package com.ds.infrastructure.hyperspace.monitor;

import com.ds.infrastructure.hyperspace.monitor.config.MetricsConfig;
import com.sun.net.httpserver.HttpExchange;
import com.sun.net.httpserver.HttpHandler;
import com.sun.net.httpserver.HttpServer;
import io.prometheus.client.*;
import io.prometheus.client.exporter.common.TextFormat;
import io.prometheus.client.hotspot.DefaultExports;
import lombok.extern.slf4j.Slf4j;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.net.HttpURLConnection;
import java.net.InetSocketAddress;
import java.net.URLDecoder;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.zip.GZIPOutputStream;

import static com.ds.infrastructure.hyperspace.monitor.EndpointType.GATEWAY;
import static com.ds.infrastructure.hyperspace.monitor.EndpointType.TARGET;

/**
 * @author  Weishaoying
 */
@Slf4j
public class PrometheusMetricsManager implements MetricsManager {

    private static final String CONNECTION_METRICS = "hyperspace_connections";
    private static final String HYPERSPACE_EXCEPTION_METRICS = "hyperspace_exception_total";
    private static final String PENDING_REQUEST_METRICS = "hyperspace_requests_pending";
    private static final String REQUEST_COUNTER_METRICS = "hyperspace_requests_total";
    private static final String PAYLOAD_SIZE_METRICS = "hyperspace_payload_bytes";
    private static final String REQUEST_GATEWAY_LATENCY_METRICS = "hyperspace_request_seconds";
    private static final String TARGET_NODE_STATUS_METRICS = "hyperspace_target_node_status";
    private static final String ENTRY_PENDING_REQUEST_METRICS = "hyperspace_entry_requests_pending";
    private static final String GIT_INFO_METRICS = "git_info";
    private static final String CONTAINER_CONFIG_VERSION_METRICS = "container_config_version";
    private static final String TRACER_DISCARDED_SPAN_METRICS = "tracer_discarded_span_total";
    private static final String OPENTRACING_CONFIG_METRICS = "opentracing_config";

    private static final String SERVICE_NAME_TAG = "service";
    private static final String ENDPOINT_TYPE_TAG = "endpoint_type";
    private static final String GATEWAY_STATUS_TAG = "gateway_status";
    private static final String TARGET_STATUS_TAG = "target_status";
    private static final String TARGET_NODE_TAG = "target_node";
    private static final String ENTRY_PORT_TAG = "entry_port";
    private static final String GIT_SHORT_COMMIT_ID_TAG = "short_commit_id";
    private static final String OPENTRACING_ENABLE_TAG = "enable";
    private static final String OPENTRACING_URL_TAG = "url";
    private static final String OPENTRACING_SAMPLE_RATE_TAG = "sample_rate";

    private final HttpServer monitorHttpServer;

    private final Gauge connectionGauge;

    private final Counter exceptionCounter;

    private final Gauge pendingRequestGauge;

    private final Counter requestCounter;

    private final Counter payloadCounter;

    private final Histogram requestLatencyHistogram;

    private final Gauge targetNodeStatusGauge;

    private final Gauge entryPendingRequestGauge;

    private final Gauge gitShortCommitIdGauge;

    private final Gauge containerConfigVersionGauge;

    private final Counter tracerDiscardedSpanCounter;

    private final Gauge opentracingConfigGauge;

    public PrometheusMetricsManager(MetricsConfig metricsConfig,
            ExecutorService executorService) throws Exception {
        this.connectionGauge = Gauge.build().name(CONNECTION_METRICS)
                .help("Hyperspace total connections.")
                .labelNames(SERVICE_NAME_TAG, ENDPOINT_TYPE_TAG)
                .register();
        this.requestCounter = Counter.build().name(REQUEST_COUNTER_METRICS)
                .help("Hyperspace total requests.")
                .labelNames(SERVICE_NAME_TAG, GATEWAY_STATUS_TAG, TARGET_STATUS_TAG)
                .register();
        this.exceptionCounter = Counter.build().name(HYPERSPACE_EXCEPTION_METRICS)
                .help("Hyperspace total errors.")
                .labelNames(SERVICE_NAME_TAG, ENDPOINT_TYPE_TAG)
                .register();
        this.requestLatencyHistogram = Histogram.build()
                .buckets(0.1, 0.5, 1, 5)
                .name(REQUEST_GATEWAY_LATENCY_METRICS)
                .help("Hyperspace request latency in seconds")
                .labelNames(SERVICE_NAME_TAG, ENDPOINT_TYPE_TAG)
                .register();

        this.pendingRequestGauge = Gauge.build().name(PENDING_REQUEST_METRICS)
                .help("Hyperspace target service total pending requests.")
                .labelNames(SERVICE_NAME_TAG)
                .register();
        this.payloadCounter = Counter.build().name(PAYLOAD_SIZE_METRICS)
                .help("Hyperspace payload bytes.")
                .labelNames(SERVICE_NAME_TAG)
                .register();

        this.targetNodeStatusGauge = Gauge.build().name(TARGET_NODE_STATUS_METRICS)
                .help("Hyperspace target service node status.")
                .labelNames(SERVICE_NAME_TAG, TARGET_NODE_TAG)
                .register();

        this.entryPendingRequestGauge = Gauge.build().name(ENTRY_PENDING_REQUEST_METRICS)
                .help("Hyperspace entry endpoint total pending requests.")
                .labelNames(ENTRY_PORT_TAG)
                .register();
        this.gitShortCommitIdGauge = Gauge.build().name(GIT_INFO_METRICS)
                .help("Hyperspace current git short commit id.")
                .labelNames(GIT_SHORT_COMMIT_ID_TAG)
                .register();
        this.containerConfigVersionGauge = Gauge.build().name(CONTAINER_CONFIG_VERSION_METRICS)
                .help("Hyperspace current running container config version.")
                .register();
        this.tracerDiscardedSpanCounter = Counter.build().name(TRACER_DISCARDED_SPAN_METRICS)
                .help("Opentracing total discarded  span.")
                .register();
        this.opentracingConfigGauge = Gauge.build().name(OPENTRACING_CONFIG_METRICS)
                .help("Hyperspace current opentracing config.")
                .labelNames(OPENTRACING_URL_TAG, OPENTRACING_ENABLE_TAG, OPENTRACING_SAMPLE_RATE_TAG)
                .register();

        this.monitorHttpServer = HttpServer.create();
        init(metricsConfig, executorService);
    }

    @Override
    public void init(MetricsConfig metricsConfig, ExecutorService executorService) throws Exception {
        DefaultExports.initialize();
        monitorHttpServer.bind(new InetSocketAddress(metricsConfig.getPort()),
                metricsConfig.getConnectionBacklogSize());
        HttpHandler httpMetricHandler = new HttpMetricHandler(CollectorRegistry.defaultRegistry);
        monitorHttpServer.createContext("/", httpMetricHandler);
        monitorHttpServer.createContext("/metrics", httpMetricHandler);

        monitorHttpServer.setExecutor(executorService);
        executorService.submit(() -> {
            monitorHttpServer.start();
            log.info("The monitor server started successfully! bind to port [{}]", metricsConfig.getPort());
        });
    }

    @Override
    public void shutdown() {
        if (monitorHttpServer != null) {
            monitorHttpServer.stop(0);
        }
    }

    @Override
    public void incrementConnectionGauge(String serviceName, EndpointType endpointType) {
        connectionGauge.labels(serviceName, endpointType.getEndpointTypeName()).inc();
    }

    @Override
    public void decrementConnectionGauge(String serviceName, EndpointType endpointType) {
        connectionGauge.labels(serviceName, endpointType.getEndpointTypeName()).dec();
    }

    @Override
    public void incrementHyperspaceException(String serviceName, EndpointType endpointType) {
        exceptionCounter.labels(serviceName, endpointType.getEndpointTypeName()).inc();
    }

    @Override
    public void incrementRequestPendingGauge(String serviceName) {
        pendingRequestGauge.labels(serviceName).inc();
    }

    @Override
    public void decrementRequestPendingGauge(String serviceName) {
        pendingRequestGauge.labels(serviceName).dec();
    }

    @Override
    public void changeTargetNodeStatusGauge(String serviceName, String nodeInfo, int status) {
        targetNodeStatusGauge.labels(serviceName, nodeInfo).set(status);
    }

    @Override
    public void removeTargetNodeStatusGauge(String serviceName, Set<String> retainTargetNodeSet) {
        List<Collector.MetricFamilySamples> collect = targetNodeStatusGauge.collect();
        for (Collector.MetricFamilySamples metricFamilySamples : collect) {
            for (Collector.MetricFamilySamples.Sample sample : metricFamilySamples.samples) {
                if (sample.labelValues.size() < 2) {
                    continue;
                }
                String serviceNameLabelValue = sample.labelValues.get(0);
                String targetNodeLabelValue = sample.labelValues.get(1);
                if (serviceNameLabelValue.equals(serviceName) && !retainTargetNodeSet.contains(targetNodeLabelValue)) {
                    targetNodeStatusGauge.remove(serviceNameLabelValue, targetNodeLabelValue);
                }
            }
        }
    }

    @Override
    public void incrementRequestCounter(String serviceName, String gatewayStatus, String targetStatus) {
        requestCounter.labels(serviceName, gatewayStatus, targetStatus).inc();
    }

    @Override
    public void changeRequestLatencyHistogram(String serviceName, long gatewayLatency, long targetLatency) {
        requestLatencyHistogram.labels(serviceName, GATEWAY.getEndpointTypeName())
                .observe(TimeUnit.MILLISECONDS.toSeconds(gatewayLatency));
        requestLatencyHistogram.labels(serviceName, TARGET.getEndpointTypeName())
                .observe(TimeUnit.MILLISECONDS.toSeconds(targetLatency));
    }

    @Override
    public void incrementPayloadCounter(String serviceName, int amount) {
        payloadCounter.labels(serviceName).inc(amount);
    }

    @Override
    public void incrementEntryRequestPendingGauge(String entryPort) {
        entryPendingRequestGauge.labels(entryPort).inc();
    }

    @Override
    public void decrementEntryRequestPendingGauge(String entryPort) {
        entryPendingRequestGauge.labels(entryPort).dec();
    }

    @Override
    public void gitShortCommitIdGauge(String shortCommitId) {
        gitShortCommitIdGauge.labels(shortCommitId).set(1);
    }

    @Override
    public void containerConfigVersionGauge(int version) {
        containerConfigVersionGauge.set(version);
    }

    @Override
    public void incrementTracerDiscardedSpanCounter() {
        tracerDiscardedSpanCounter.inc();
    }

    @Override
    public void opentracingConfigGauge(String url, String enable, String sampleRate) {
        opentracingConfigGauge.clear();
        if (url == null) {
            url = "";
        }
        opentracingConfigGauge.labels(url, enable, sampleRate).set(1);
    }

    static class HttpMetricHandler implements HttpHandler {

        private CollectorRegistry registry;
        private final LocalByteArray response = new LocalByteArray();

        HttpMetricHandler(CollectorRegistry registry) {
            this.registry = registry;
        }

        @Override
        public void handle(HttpExchange t) throws IOException {
            String query = t.getRequestURI().getRawQuery();
            ByteArrayOutputStream response = this.response.get();
            response.reset();
            OutputStreamWriter osw = new OutputStreamWriter(response);
            TextFormat.write004(osw,
                    registry.filteredMetricFamilySamples(parseQuery(query)));
            osw.flush();
            osw.close();
            response.flush();
            response.close();

            t.getResponseHeaders().set("Content-Type",
                    TextFormat.CONTENT_TYPE_004);
            if (shouldUseCompression(t)) {
                t.getResponseHeaders().set("Content-Encoding", "gzip");
                t.sendResponseHeaders(HttpURLConnection.HTTP_OK, 0);
                final GZIPOutputStream os = new GZIPOutputStream(t.getResponseBody());
                response.writeTo(os);
                os.close();
            } else {
                t.getResponseHeaders().set("Content-Length",
                        String.valueOf(response.size()));
                t.sendResponseHeaders(HttpURLConnection.HTTP_OK, response.size());
                response.writeTo(t.getResponseBody());
            }
            t.close();
        }

    }

    protected static boolean shouldUseCompression(HttpExchange exchange) {
        List<String> encodingHeaders = exchange.getRequestHeaders().get("Accept-Encoding");
        if (encodingHeaders == null) {
            return false;
        }

        for (String encodingHeader : encodingHeaders) {
            String[] encodings = encodingHeader.split(",");
            for (String encoding : encodings) {
                if (encoding.trim().toLowerCase().equals("gzip")) {
                    return true;
                }
            }
        }
        return false;
    }

    protected static Set<String> parseQuery(String query) throws IOException {
        Set<String> names = new HashSet<String>();
        if (query != null) {
            String[] pairs = query.split("&");
            for (String pair : pairs) {
                int idx = pair.indexOf("=");
                if (idx != -1 && URLDecoder.decode(pair.substring(0, idx), "UTF-8").equals("name[]")) {
                    names.add(URLDecoder.decode(pair.substring(idx + 1), "UTF-8"));
                }
            }
        }
        return names;
    }

    private static class LocalByteArray extends ThreadLocal<ByteArrayOutputStream> {
        @Override
        protected ByteArrayOutputStream initialValue() {
            return new ByteArrayOutputStream(1 << 20);
        }
    }

}
